package com.luoqifei.yomob.realtime.analysis.kafka.stream;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;

import java.util.Properties;

public class KstreamQuery {
    public static void main(String[] args) {
        Properties config = new Properties();
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application");
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        config.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);

        KStreamBuilder builder = new KStreamBuilder();
        KafkaStreams streams = new KafkaStreams(builder, config);
        streams.start();
        ReadOnlyKeyValueStore<String,Long> keyValueStore = streams.store("Counts", QueryableStoreTypes.keyValueStore());
        System.out.println("count for hello:"+keyValueStore.get("hello"));
        KeyValueIterator<String,Long> range  = keyValueStore.all();
        while (range.hasNext()){
            KeyValue<String,Long> next = range.next();
            System.out.println("count for "+next.key+ " : "+ next.value);
        }
    }
}
